package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * spark的运行过程中如果出现了相同的键被拉取到对应的分区，这个过程称之为shuffle
 * 注：spark的shuffle和mapreduce的shuffle原理是一样，都是要进行落盘
 *
 * RDD: 弹性的分布式数据集
 * 弹性：RDD将来在计算的时候，其中的数据可以是很大，也可以是很小
 * 分布式：数据可以分布在多台服务器中，RDD中的分区来自于block块，而今后的block块会来自不同的datanode
 * 数据集：RDD自身是不存储数据的，只是一个代码计算逻辑，今后触发作业执行的时候，数据会在RDD之间流动
 *
 * 5大特性：（面试必会的！！）
 * 1）RDD是由一些分区构成的 读取文件时有多少个block块，RDD中就会有多少个分区
 * 注：默认情况下，所有的RDD中的分区数是一样的，无论是shuffle之前还是shuffle之后的,在最开始加载数据的时候决定的
 *
 * 2）函数实际上是作用在RDD中的分区上的，一个分区是由一个task处理，有多少个分区，总共就有多少个task
 * 注：函数在spark中称之为算子（转换transformation算子  RDD-->RDD，行动action算子 RDD->Other数据类型）
 *
 * 3）RDD之间存在一些依赖关系，后一个RDD中的数据是依赖与前一个RDD的计算结果，数据像水流一样在RDD之间流动
 * 注：
 * 3.1 RDD之间有两种依赖关系
 *        a. 窄依赖  后一个RDD中分区数据对应前一个RDD中的一个分区数据 1对1的关系
 *           b. 宽依赖  后一个RDD中分区数据来自于前一个RDD中的多个分区数据  1对多的关系  shuffle
 *           3.2 因为有了依赖关系，将整个作业划分了一个一个stage阶段  sumNum(stage) = Num(宽依赖) + 1
 *           3.3 窄依赖的分区数是不可以改变，取决于第一个RDD分区数，宽依赖可以在产生shuffle的算子上设置分区数
 *
 * 4）分区类的算子只能作用在kv格式的RDD上，groupByKey reduceByKey
 * 5）spark为task计算提供了精确的计算位置，移动计算而不移动数据
 */
object Demo2Partition {
  def main(args: Array[String]): Unit = {
    //1、创建Spark环境
    //1.1 创建配置文件对象
    val conf: SparkConf = new SparkConf()

    //1.2 指定运行的模式（local  Standalone  Mesos  YARN）
    conf.setMaster("local") //可以执行所运行需要核数资源local[2]，不指定的话默认使用所有的资源执行程序

    //1.3 给spark作业起一个名字
    conf.setAppName("wc")

    //2、创建spark运行时的上下文对象
    val sparkContext: SparkContext = new SparkContext(conf)

    //3、读取文件数据
    //    val wordsLine: RDD[String] = sparkContext.textFile("spark/data/ws/*", minPartitions = 7)
    val wordsLine: RDD[String] = sparkContext.textFile("spark/data/ws/*")
    println(s"wordsLineRDD分区数是：${wordsLine.getNumPartitions}")

    //4、每一行根据|分隔符进行切分
    val words: RDD[String] = wordsLine.flatMap(_.split("\\|"))
    println(s"wordsRDD分区数是：${words.getNumPartitions}")

    val wordsTuple2: RDD[(String, Int)] = words.map((_, 1))
    println(s"wordsTuple2RDD分区数是：${wordsTuple2.getNumPartitions}")

    //产生shuffle的算子上可以单独设置分区数
    val wordsTuple2Group: RDD[(String, Iterable[(String, Int)])] = wordsTuple2.groupBy(_._1, 5)
    println(s"wordsTuple2GroupRDD分区数是：${wordsTuple2Group.getNumPartitions}")

    val wordCount: RDD[(String, Int)] = wordsTuple2Group.map((kv: (String, Iterable[(String, Int)])) => (kv._1, kv._2.size))
    println(s"wordCountRDD分区数是：${wordCount.getNumPartitions}")

    wordCount.saveAsTextFile("spark/data/word_count2")

  }
}
